package com.ruicar.afs.cloud.batch.frame.control;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.exceptions.ExceptionUtil;
import com.ruicar.afs.cloud.batch.entity.AfsBatchInfo;
import com.ruicar.afs.cloud.batch.enums.ErrorType;
import com.ruicar.afs.cloud.batch.enums.MessageType;
import com.ruicar.afs.cloud.batch.frame.AbsBatchRunner;
import com.ruicar.afs.cloud.batch.frame.bean.BatchDateInfo;
import com.ruicar.afs.cloud.batch.frame.bean.BatchRunStepInfo;
import com.ruicar.afs.cloud.batch.frame.logger.WsMonitorHelper;
import com.ruicar.afs.cloud.batch.service.AfsBatchInfoService;
import com.ruicar.afs.cloud.batch.service.AfsBatchLogStepService;
import lombok.extern.slf4j.Slf4j;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class BatchStepControlRunner extends BatchControlRunner {
    private final BatchRunStepInfo batchRunStepInfo;
    private final CountDownLatch superCountDownLatch;
    private final AfsBatchLogStepService afsBatchLogStepService;
    private final AfsBatchInfoService afsBatchInfoService;
    private final MonitorControl monitorControl;
    private ErrorType errorType = ErrorType.SELF;
    public BatchStepControlRunner(BatchRunStepInfo batchRunStepInfo, CountDownLatch superCountDownLatch, AfsBatchLogStepService afsBatchLogStepService, AfsBatchInfoService afsBatchInfoService, MonitorControl monitorControl) {
        this.batchRunStepInfo = batchRunStepInfo;
        this.superCountDownLatch = superCountDownLatch;
        this.afsBatchLogStepService = afsBatchLogStepService;
        this.afsBatchInfoService = afsBatchInfoService;
        this.monitorControl = monitorControl;
    }

    @Override
    public void run() {
        monitorControl.add(this);
        AfsBatchInfo afsBatchInfo =afsBatchInfoService.getSystemBatchInfoWithOutUpdate(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName());
        try {
            if (afsBatchLogStepService.hasFinish(batchRunStepInfo.getAfsBatchJobStepInfo(),afsBatchInfo.getBatchDate())&&BatchHelper.shouldRun(batchRunStepInfo.getAfsBatchJobStepInfo().getStepType(),afsBatchInfo)) {
                log.info("=========步骤[{}-{}-{}],批量日[{}]已完成处理子步骤=========",batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo(),afsBatchInfo.getBatchDate());
                setSelfFinished(true);
                runSubStep();
            } else {
                runSelfStep();
            }
         }catch (Throwable e){
            if(errorType ==ErrorType.SELF) {
                afsBatchLogStepService.saveError(batchRunStepInfo.getAfsBatchJobStepInfo(), ExceptionUtil.stacktraceToString(e, 5000), afsBatchInfo.getBatchDate());
                log.error(MessageFormat.format("=========步骤[{0}]-[{1}]-[{2}]运行异常=========",batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo()),e);
                WsMonitorHelper.sendMessage(MessageType.ERROR,MessageFormat.format("步骤[{0}]-[{1}]-[{2}]运行异常",batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo()));
                WsMonitorHelper.sendMessage(MessageType.ERROR,ExceptionUtil.stacktraceToString(e,5000));
            }
            this.setThrowable(e);
            throw new RuntimeException(e);
        }finally {
            if (superCountDownLatch!=null){
                afsBatchInfoService.cutDown(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),1,MessageFormat.format("StepRunner-{0}-{1}-{2}",afsBatchInfo.getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo()));
                superCountDownLatch.countDown();
            }
            this.setFinished();
        }
    }

    private void runSelfStep() throws Throwable{
        CountDownLatch countDownLatch = new CountDownLatch(batchRunStepInfo.getAfsBatchJobStepInfo().getThreadNum());
        Class<? extends AbsBatchRunner> runnerClass = (Class<? extends AbsBatchRunner>) Class.forName(batchRunStepInfo.getAfsBatchJobStepInfo().getRunClass());
        String params = batchRunStepInfo.getAfsBatchJobStepInfo().getStepParam();
        List<AbsBatchRunner> absBatchRunners = new ArrayList<>();
        AfsStepRunnerExceptionHandler afsStepRunnerExceptionHandler = new AfsStepRunnerExceptionHandler();
        AfsBatchInfo afsBatchInfo =afsBatchInfoService.getSystemBatchInfo(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName());
        afsBatchLogStepService.doLog(batchRunStepInfo.getAfsBatchJobStepInfo(),afsBatchInfo.getBatchDate());
        Lock lock = new ReentrantLock();
        for(int i = 0; i<batchRunStepInfo.getAfsBatchJobStepInfo().getThreadNum(); i++){
            AbsBatchRunner absBatchRunner = runnerClass.newInstance();
            absBatchRunner.setCountDownLatch(countDownLatch);
            absBatchRunner.setUncaughtExceptionHandler(afsStepRunnerExceptionHandler);
            absBatchRunner.setBatchNo(batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo());
            absBatchRunner.setParams(params);
            absBatchRunner.setStepName(batchRunStepInfo.getAfsBatchJobStepInfo().getStepName());
            absBatchRunner.setSystemName(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName());
            absBatchRunner.setStepNo(batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo());
            absBatchRunner.setThreadNum(i);
            absBatchRunner.setMonitorControl(monitorControl);
            absBatchRunner.setShareLock(lock);
            BatchDateInfo batchDateInfo = new BatchDateInfo(DateUtil.parse(afsBatchInfo.getBatchDate()),BatchHelper.isHoliday(afsBatchInfo.getBatchDate()));
            absBatchRunner.setBatchDateInfo(batchDateInfo);
            absBatchRunner.setName(MessageFormat.format("步骤线程-{0}-{1}-{2}-{3}",batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo(),i));
            absBatchRunners.add(absBatchRunner);
        }
        afsBatchInfoService.addRunThreadNum(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getThreadNum(),MessageFormat.format("StepRunner-{0}-{1}-{2}",afsBatchInfo.getSystemName(),batchRunStepInfo.getAfsBatchJobStepInfo().getBatchNo(),batchRunStepInfo.getAfsBatchJobStepInfo().getStepNo()));
        afsStepRunnerExceptionHandler.setAbsBatchRunners(absBatchRunners);
        for(AbsBatchRunner runner:absBatchRunners){
            runner.start();
        }
        countDownLatch.await();
        boolean exceptionFlag = false;
        Throwable throwable = null;
        for (AbsBatchRunner runner : absBatchRunners) {
            if(runner.getThrowable()!=null){
                if(throwable==null&&!(runner.getThrowable() instanceof InterruptedException)){
                    throwable = runner.getThrowable();
                }
                exceptionFlag = true;
            }
        }
        setSelfFinished(true);
        if(!exceptionFlag) {
            afsBatchLogStepService.finished(batchRunStepInfo.getAfsBatchJobStepInfo(),afsBatchInfo.getBatchDate());
            runSubStep();
        }else{
            afsBatchLogStepService.saveError(batchRunStepInfo.getAfsBatchJobStepInfo(), ExceptionUtil.stacktraceToString(throwable,5000),afsBatchInfo.getBatchDate());
            throw throwable;
        }
    }

    private void runSubStep() throws Throwable{
        if(batchRunStepInfo.getSubBatchSteps().size()>0) {
            CountDownLatch countDownLatch = new CountDownLatch(batchRunStepInfo.getSubBatchSteps().size());
            List<BatchStepControlRunner> stepControlRunners = new ArrayList<>();
            for (BatchRunStepInfo subBatchStep : batchRunStepInfo.getSubBatchSteps()) {
                afsBatchInfoService.addRunThreadNum(batchRunStepInfo.getAfsBatchJobStepInfo().getSystemName(),1,MessageFormat.format("StepRunner-{0}-{1}-{2}",subBatchStep.getAfsBatchJobStepInfo().getSystemName(),subBatchStep.getAfsBatchJobStepInfo().getBatchNo(),subBatchStep.getAfsBatchJobStepInfo().getStepNo()));
                BatchStepControlRunner batchStepControlRunner = new BatchStepControlRunner(subBatchStep, countDownLatch, afsBatchLogStepService, afsBatchInfoService,monitorControl);
                batchStepControlRunner.setDaemon(true);
                batchStepControlRunner.setName(MessageFormat.format("批量步骤-{0}-{1}-{2}",subBatchStep.getAfsBatchJobStepInfo().getSystemName(),subBatchStep.getAfsBatchJobStepInfo().getBatchNo(),subBatchStep.getAfsBatchJobStepInfo().getStepName()));
                stepControlRunners.add(batchStepControlRunner);
                batchStepControlRunner.start();
            }
            countDownLatch.await();
            setSubFinished(true);
            for(BatchStepControlRunner batchStepControlRunner:stepControlRunners){
                if(batchStepControlRunner.getThrowable()!=null){
                    errorType = ErrorType.SUB;
                    log.error("========={}执行异常==========",batchStepControlRunner.getName());
                    throw batchStepControlRunner.getThrowable();
                }
            }
        }
    }
}
